{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3cee6459-5776-4200-b073-1da687f062ac",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "94873843-9bf0-4135-8e82-e1bd647279b6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===== 启动用户信息和行为表持续更新程序 =====\n",
      "目标数据库: soo\n",
      "提示: 按 Ctrl+C 可停止所有表的更新操作，输入 'pause' 暂停，'resume' 恢复\n",
      "\n",
      "✅ 准备在现有表上追加数据，不会创建或替换原有表\n",
      "[2025-08-12 18:35:54] 启动 customers_info 表更新线程，间隔 1 秒，每次新增 10 条数据\n",
      "[2025-08-12 18:35:54] 启动 user_behavior_demo 表更新线程，间隔 1 秒，每次新增 15 条数据\n",
      "[2025-08-12 18:36:09] user_behavior_demo 表新增 15 条数据，当前总数据量: 23362 条\n",
      "[2025-08-12 18:36:10] customers_info 表新增 10 条数据，当前总数据量: 800010 条\n",
      "[2025-08-12 18:36:26] user_behavior_demo 表新增 15 条数据，当前总数据量: 23400 条\n",
      "[2025-08-12 18:36:28] customers_info 表新增 10 条数据，当前总数据量: 800020 条\n",
      "[2025-08-12 18:36:44] user_behavior_demo 表新增 15 条数据，当前总数据量: 23435 条\n",
      "[2025-08-12 18:36:45] customers_info 表新增 10 条数据，当前总数据量: 800030 条\n",
      "[2025-08-12 18:37:01] customers_info 表新增 10 条数据，当前总数据量: 800040 条\n",
      "[2025-08-12 18:37:04] user_behavior_demo 表新增 15 条数据，当前总数据量: 23473 条\n",
      "[2025-08-12 18:37:17] customers_info 表新增 10 条数据，当前总数据量: 800050 条\n",
      "[2025-08-12 18:37:21] user_behavior_demo 表新增 15 条数据，当前总数据量: 23506 条\n",
      "[2025-08-12 18:37:32] customers_info 表新增 10 条数据，当前总数据量: 800060 条\n"
     ]
    },
    {
     "name": "stdin",
     "output_type": "stream",
     "text": [
      ">  pause\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2025-08-12 18:37:34] 所有表暂停更新\n",
      "[2025-08-12 18:37:38] user_behavior_demo 表新增 15 条数据，当前总数据量: 23540 条\n",
      "[2025-08-12 18:37:38] user_behavior_demo 表暂停，等待恢复...\n",
      "[2025-08-12 18:37:44] customers_info 表新增 10 条数据，当前总数据量: 800070 条\n",
      "[2025-08-12 18:37:44] customers_info 表暂停，等待恢复...\n"
     ]
    },
    {
     "name": "stdin",
     "output_type": "stream",
     "text": [
      ">  resume\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2025-08-12 18:55:41] 所有表恢复更新[2025-08-12 18:55:41] user_behavior_demo 表恢复更新\n",
      "\n",
      "[2025-08-12 18:55:41] customers_info 表恢复更新\n",
      "[2025-08-12 18:56:03] customers_info 表新增 10 条数据，当前总数据量: 800080 条\n",
      "[2025-08-12 18:56:06] user_behavior_demo 表新增 15 条数据，当前总数据量: 23580 条\n",
      "[2025-08-12 18:56:29] customers_info 表新增 10 条数据，当前总数据量: 800090 条\n",
      "[2025-08-12 18:56:35] user_behavior_demo 表新增 15 条数据，当前总数据量: 23617 条\n",
      "[2025-08-12 18:56:49] customers_info 表新增 10 条数据，当前总数据量: 800100 条\n",
      "[2025-08-12 18:56:56] user_behavior_demo 表新增 15 条数据，当前总数据量: 23652 条\n",
      "[2025-08-12 18:57:11] customers_info 表新增 10 条数据，当前总数据量: 800110 条\n",
      "[2025-08-12 18:57:23] user_behavior_demo 表新增 15 条数据，当前总数据量: 23687 条\n",
      "[2025-08-12 18:57:35] customers_info 表新增 10 条数据，当前总数据量: 800120 条\n",
      "[2025-08-12 18:57:45] user_behavior_demo 表新增 15 条数据，当前总数据量: 23724 条\n",
      "[2025-08-12 18:57:55] customers_info 表新增 10 条数据，当前总数据量: 800130 条\n",
      "[2025-08-12 18:58:06] user_behavior_demo 表新增 15 条数据，当前总数据量: 23765 条\n",
      "[2025-08-12 18:58:17] customers_info 表新增 10 条数据，当前总数据量: 800140 条\n",
      "[2025-08-12 18:58:36] user_behavior_demo 表新增 15 条数据，当前总数据量: 23797 条\n",
      "[2025-08-12 18:58:41] customers_info 表新增 10 条数据，当前总数据量: 800150 条\n",
      "[2025-08-12 18:58:58] user_behavior_demo 表新增 15 条数据，当前总数据量: 23831 条\n",
      "[2025-08-12 18:59:03] customers_info 表新增 10 条数据，当前总数据量: 800160 条\n",
      "[2025-08-12 18:59:27] customers_info 表新增 10 条数据，当前总数据量: 800170 条\n",
      "[2025-08-12 18:59:31] user_behavior_demo 表新增 15 条数据，当前总数据量: 23864 条\n"
     ]
    },
    {
     "name": "stdin",
     "output_type": "stream",
     "text": [
      ">  pause\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2025-08-12 18:59:42] 所有表暂停更新\n",
      "[2025-08-12 18:59:49] customers_info 表新增 10 条数据，当前总数据量: 800180 条\n",
      "[2025-08-12 18:59:49] customers_info 表暂停，等待恢复...\n",
      "[2025-08-12 18:59:51] user_behavior_demo 表新增 15 条数据，当前总数据量: 23901 条\n",
      "[2025-08-12 18:59:51] user_behavior_demo 表暂停，等待恢复...\n",
      "\n",
      "所有表更新线程已手动停止\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import random\n",
    "import time\n",
    "import threading\n",
    "from datetime import datetime, timedelta\n",
    "from sqlalchemy import create_engine\n",
    "from sqlalchemy.exc import SQLAlchemyError\n",
    "\n",
    "# --------------------------\n",
    "# 1. 数据库配置\n",
    "# --------------------------\n",
    "DB_CONFIG = {\n",
    "    \"user\": \"root\",\n",
    "    \"password\": \"526108\",\n",
    "    \"host\": \"localhost\",\n",
    "    \"port\": 3306,\n",
    "    \"database\": \"soo\"\n",
    "}\n",
    "\n",
    "# 创建数据库连接引擎\n",
    "engine = create_engine(\n",
    "    f\"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}\",\n",
    "    pool_size=15,\n",
    "    max_overflow=5,\n",
    "    pool_recycle=3600,\n",
    "    connect_args={\"connect_timeout\": 1000}\n",
    ")\n",
    "\n",
    "# --------------------------\n",
    "# 2. 通用工具函数\n",
    "# --------------------------\n",
    "def generate_unique_ids(prefix, num, existing_ids):\n",
    "    new_ids = []\n",
    "    while len(new_ids) < num:\n",
    "        candidate_ids = [f\"{prefix}_{random.randint(100000, 999999)}\" for _ in range(num * 10)]\n",
    "        valid_ids = [cid for cid in candidate_ids if cid not in existing_ids and cid not in new_ids]\n",
    "        new_ids.extend(valid_ids[:num - len(new_ids)])\n",
    "    return new_ids\n",
    "\n",
    "def get_existing_ids(table_name, id_column):\n",
    "    try:\n",
    "        query = f\"SELECT {id_column} FROM {table_name}\"\n",
    "        existing_data = pd.read_sql(query, con=engine)\n",
    "        return set(existing_data[id_column].tolist())\n",
    "    except (SQLAlchemyError, ValueError):\n",
    "        return set()\n",
    "\n",
    "# --------------------------\n",
    "# 3. 省份城市数据\n",
    "# --------------------------\n",
    "cities = {\n",
    "   \"北京市\": [\"东城区\", \"西城区\", \"朝阳区\", \"海淀区\", \"丰台区\", \"石景山区\", \"通州区\", \"顺义区\", \"大兴区\", \"昌平区\", \"怀柔区\", \"平谷区\", \"门头沟区\", \"密云区\", \"延庆区\"],\n",
    "    \"天津市\": [\"和平区\", \"河东区\", \"河西区\", \"南开区\", \"河北区\", \"红桥区\", \"东丽区\", \"西青区\", \"津南区\", \"北辰区\", \"武清区\", \"宝坻区\", \"滨海新区\", \"宁河区\", \"静海区\", \"蓟州区\"],\n",
    "    \"上海市\": [\"黄浦区\", \"徐汇区\", \"长宁区\", \"静安区\", \"普陀区\", \"虹口区\", \"杨浦区\", \"闵行区\", \"宝山区\", \"嘉定区\", \"浦东新区\", \"金山区\", \"松江区\", \"青浦区\", \"奉贤区\", \"崇明区\"],\n",
    "    \"重庆市\": [\"渝中区\", \"大渡口区\", \"江北区\", \"沙坪坝区\", \"九龙坡区\", \"南岸区\", \"北碚区\", \"渝北区\", \"巴南区\", \"长寿区\", \"璧山区\", \"大足区\", \"荣昌区\", \"铜梁区\", \"合川区\", \"永川区\", \"江津区\", \"綦江区\", \"潼南区\", \"万州区\", \"涪陵区\", \"黔江区\"],\n",
    "    \"河北省\": [\"石家庄市\", \"唐山市\", \"秦皇岛市\", \"邯郸市\", \"邢台市\", \"保定市\", \"张家口市\", \"承德市\", \"沧州市\", \"廊坊市\", \"衡水市\"],\n",
    "    \"山西省\": [\"太原市\", \"大同市\", \"阳泉市\", \"长治市\", \"晋城市\", \"朔州市\", \"晋中市\", \"运城市\", \"忻州市\", \"临汾市\", \"吕梁市\"],\n",
    "    \"辽宁省\": [\"沈阳市\", \"大连市\", \"鞍山市\", \"抚顺市\", \"本溪市\", \"丹东市\", \"锦州市\", \"营口市\", \"阜新市\", \"辽阳市\", \"盘锦市\", \"铁岭市\", \"朝阳市\", \"葫芦岛市\"],\n",
    "    \"吉林省\": [\"长春市\", \"吉林市\", \"四平市\", \"辽源市\", \"通化市\", \"白山市\", \"松原市\", \"白城市\", \"延边朝鲜族自治州\"],\n",
    "    \"黑龙江省\": [\"哈尔滨市\", \"齐齐哈尔市\", \"鸡西市\", \"鹤岗市\", \"双鸭山市\", \"大庆市\", \"伊春市\", \"佳木斯市\", \"七台河市\", \"牡丹江市\", \"黑河市\", \"绥化市\", \"大兴安岭地区\"],\n",
    "    \"江苏省\": [\"南京市\", \"无锡市\", \"徐州市\", \"常州市\", \"苏州市\", \"南通市\", \"连云港市\", \"淮安市\", \"盐城市\", \"扬州市\", \"镇江市\", \"泰州市\", \"宿迁市\"],\n",
    "    \"浙江省\": [\"杭州市\", \"宁波市\", \"温州市\", \"嘉兴市\", \"湖州市\", \"绍兴市\", \"金华市\", \"衢州市\", \"舟山市\", \"台州市\", \"丽水市\"],\n",
    "    \"安徽省\": [\"合肥市\", \"芜湖市\", \"蚌埠市\", \"淮南市\", \"马鞍山市\", \"淮北市\", \"铜陵市\", \"安庆市\", \"黄山市\", \"滁州市\", \"阜阳市\", \"宿州市\", \"六安市\", \"亳州市\", \"池州市\", \"宣城市\"],\n",
    "    \"福建省\": [\"福州市\", \"厦门市\", \"莆田市\", \"三明市\", \"泉州市\", \"漳州市\", \"南平市\", \"龙岩市\", \"宁德市\"],\n",
    "    \"江西省\": [\"南昌市\", \"景德镇市\", \"萍乡市\", \"九江市\", \"新余市\", \"鹰潭市\", \"赣州市\", \"吉安市\", \"宜春市\", \"抚州市\", \"上饶市\"],\n",
    "    \"山东省\": [\"济南市\", \"青岛市\", \"淄博市\", \"枣庄市\", \"东营市\", \"烟台市\", \"潍坊市\", \"济宁市\", \"泰安市\", \"威海市\", \"日照市\", \"莱芜市\", \"临沂市\", \"德州市\", \"聊城市\", \"滨州市\", \"菏泽市\"],\n",
    "    \"河南省\": [\"郑州市\", \"开封市\", \"洛阳市\", \"平顶山市\", \"安阳市\", \"鹤壁市\", \"新乡市\", \"焦作市\", \"濮阳市\", \"许昌市\", \"漯河市\", \"三门峡市\", \"南阳市\", \"商丘市\", \"信阳市\", \"周口市\", \"驻马店市\", \"济源市\"],\n",
    "    \"湖北省\": [\"武汉市\", \"黄石市\", \"十堰市\", \"宜昌市\", \"襄阳市\", \"鄂州市\", \"荆门市\", \"孝感市\", \"荆州市\", \"黄冈市\", \"咸宁市\", \"随州市\", \"恩施土家族苗族自治州\"],\n",
    "    \"湖南省\": [\"长沙市\", \"株洲市\", \"湘潭市\", \"衡阳市\", \"邵阳市\", \"岳阳市\", \"常德市\", \"张家界市\", \"益阳市\", \"郴州市\", \"永州市\", \"怀化市\", \"娄底市\", \"湘西土家族苗族自治州\"],\n",
    "    \"广东省\": [\"广州市\", \"深圳市\", \"珠海市\", \"汕头市\", \"韶关市\", \"佛山市\", \"江门市\", \"湛江市\", \"茂名市\", \"肇庆市\", \"惠州市\", \"梅州市\", \"汕尾市\", \"河源市\", \"阳江市\", \"清远市\", \"东莞市\", \"中山市\", \"潮州市\", \"揭阳市\", \"云浮市\"],\n",
    "    \"海南省\": [\"海口市\", \"三亚市\", \"三沙市\", \"儋州市\", \"文昌市\", \"琼海市\", \"万宁市\", \"东方市\", \"定安县\", \"屯昌县\", \"澄迈县\", \"临高县\", \"白沙黎族自治县\", \"昌江黎族自治县\", \"乐东黎族自治县\", \"陵水黎族自治县\", \"保亭黎族苗族自治县\", \"琼中黎族苗族自治县\"],\n",
    "    \"四川省\": [\"成都市\", \"自贡市\", \"攀枝花市\", \"泸州市\", \"德阳市\", \"绵阳市\", \"广元市\", \"遂宁市\", \"内江市\", \"乐山市\", \"南充市\", \"眉山市\", \"宜宾市\", \"广安市\", \"达州市\", \"雅安市\", \"巴中市\", \"资阳市\", \"阿坝藏族羌族自治州\", \"甘孜藏族自治州\", \"凉山彝族自治州\"],\n",
    "    \"贵州省\": [\"贵阳市\", \"六盘水市\", \"遵义市\", \"安顺市\", \"毕节市\", \"铜仁市\", \"黔西南布依族苗族自治州\", \"黔东南苗族侗族自治州\", \"黔南布依族苗族自治州\"],\n",
    "    \"云南省\": [\"昆明市\", \"曲靖市\", \"玉溪市\", \"保山市\", \"昭通市\", \"丽江市\", \"普洱市\", \"临沧市\", \"楚雄彝族自治州\", \"红河哈尼族彝族自治州\", \"文山壮族苗族自治州\", \"西双版纳傣族自治州\", \"大理白族自治州\", \"德宏傣族景颇族自治州\", \"怒江傈僳族自治州\", \"迪庆藏族自治州\"],\n",
    "    \"陕西省\": [\"西安市\", \"铜川市\", \"宝鸡市\", \"咸阳市\", \"渭南市\", \"延安市\", \"汉中市\", \"榆林市\", \"安康市\", \"商洛市\"],\n",
    "    \"甘肃省\": [\"兰州市\", \"嘉峪关市\", \"金昌市\", \"白银市\", \"天水市\", \"武威市\", \"张掖市\", \"平凉市\", \"酒泉市\", \"庆阳市\", \"定西市\", \"陇南市\", \"临夏回族自治州\", \"甘南藏族自治州\"],\n",
    "    \"青海省\": [\"西宁市\", \"海东市\", \"海北藏族自治州\", \"黄南藏族自治州\", \"海南藏族自治州\", \"果洛藏族自治州\", \"玉树藏族自治州\", \"海西蒙古族藏族自治州\"],\n",
    "    \"台湾省\": [\"台北市\", \"高雄市\", \"台中市\", \"台南市\", \"新北市\", \"桃园市\", \"新竹市\", \"基隆市\", \"嘉义市\"],\n",
    "    \"内蒙古自治区\": [\"呼和浩特市\", \"包头市\", \"乌海市\", \"赤峰市\", \"通辽市\", \"鄂尔多斯市\", \"呼伦贝尔市\", \"巴彦淖尔市\", \"乌兰察布市\", \"兴安盟\", \"锡林郭勒盟\", \"阿拉善盟\"],\n",
    "    \"广西壮族自治区\": [\"南宁市\", \"柳州市\", \"桂林市\", \"梧州市\", \"北海市\", \"防城港市\", \"钦州市\", \"贵港市\", \"玉林市\", \"百色市\", \"贺州市\", \"河池市\", \"来宾市\", \"崇左市\"],\n",
    "    \"西藏自治区\": [\"拉萨市\", \"日喀则市\", \"昌都市\", \"林芝市\", \"山南市\", \"那曲市\", \"阿里地区\"],\n",
    "    \"宁夏回族自治区\": [\"银川市\", \"石嘴山市\", \"吴忠市\", \"固原市\", \"中卫市\"],\n",
    "    \"新疆维吾尔自治区\": [\"乌鲁木齐市\", \"克拉玛依市\", \"吐鲁番市\", \"哈密市\", \"喀什市\", \"阿克苏市\", \"和田市\", \"昌吉回族自治州\", \"博尔塔拉蒙古自治州\", \"巴音郭楞蒙古自治州\", \"克孜勒苏柯尔克孜自治州\", \"伊犁哈萨克自治州\", \"塔城地区\", \"阿勒泰地区\"],\n",
    "    \"香港特别行政区\": [\"中西区\", \"湾仔区\", \"东区\", \"南区\", \"油尖旺区\", \"深水埗区\", \"九龙城区\", \"黄大仙区\", \"观塘区\", \"荃湾区\", \"屯门区\", \"元朗区\", \"北区\", \"大埔区\", \"西贡区\", \"沙田区\", \"葵青区\", \"离岛区\"],\n",
    "    \"澳门特别行政区\": [\"花地玛堂区\", \"圣安多尼堂区\", \"大堂区\", \"望德堂区\", \"风顺堂区\", \"嘉模堂区\", \"路凼填海区\", \"圣方济各堂区\"]\n",
    "}\n",
    "\n",
    "cities_flat = []\n",
    "for province, city_list in cities.items():\n",
    "    for city in city_list:\n",
    "        cities_flat.append(city)\n",
    "\n",
    "# --------------------------\n",
    "# 4. 数据生成函数\n",
    "# --------------------------\n",
    "def generate_customers_info(num_new=10):\n",
    "    existing_user_ids = get_existing_ids(\"customers_info\", \"用户ID\")\n",
    "    user_ids = generate_unique_ids(\"CUST\", num_new, existing_user_ids)\n",
    "\n",
    "    ages = []\n",
    "    for _ in range(num_new):\n",
    "        if random.random() < 0.02:\n",
    "            age = random.choice([random.randint(0, 10), random.randint(101, 150)])\n",
    "        else:\n",
    "            age = int(np.random.normal(40, 10))\n",
    "            age = max(18, min(65, age))\n",
    "        ages.append(age)\n",
    "\n",
    "    genders = []\n",
    "    for _ in range(num_new):\n",
    "        if random.random() < 0.01:\n",
    "            genders.append(None)\n",
    "        else:\n",
    "            genders.append(random.choice([\"男\", \"女\"]))\n",
    "\n",
    "    rights = random.choices([\"普通\", \"会员\"], weights=[0.7, 0.3], k=num_new)\n",
    "    \n",
    "    start_date = datetime.strptime(\"2022-01-01\", \"%Y-%m-%d\")\n",
    "    reg_dates = [\n",
    "        (start_date + timedelta(days=random.randint(0, 730))).strftime(\"%Y-%m-%d\") \n",
    "        for _ in range(num_new)\n",
    "    ]\n",
    "    \n",
    "    verify_status = random.choices([\"是\", \"否\"], weights=[0.8, 0.2], k=num_new)\n",
    "    \n",
    "    locations = []\n",
    "    for _ in range(num_new):\n",
    "        if random.random() < 0.02:\n",
    "            locations.append(\"未知\")\n",
    "        else:\n",
    "            locations.append(random.choice(cities_flat))\n",
    "\n",
    "    return pd.DataFrame({\n",
    "        \"用户ID\": user_ids,\n",
    "        \"用户所在地\": locations,\n",
    "        \"年龄\": ages,\n",
    "        \"性别\": genders,\n",
    "        \"权益\": rights,\n",
    "        \"注册时间\": reg_dates,\n",
    "        \"是否实名认证\": verify_status\n",
    "    })\n",
    "\n",
    "def generate_user_behavior(num_new=15):\n",
    "    existing_user_ids = get_existing_ids(\"customers_info\", \"用户ID\")\n",
    "    if not existing_user_ids:\n",
    "        existing_user_ids = {f\"CUST_{random.randint(100000, 999999)}\" for _ in range(20)}\n",
    "    \n",
    "    product_ids = [f\"PROD_{random.randint(100000, 999999)}\" for _ in range(50)]\n",
    "    \n",
    "    behavior_data = {\n",
    "        \"用户ID\": [], \"商品ID\": [], \"时间\": [], \"行为\": [], \"备注\": []\n",
    "    }\n",
    "    \n",
    "    start_date = datetime(2023, 1, 1)\n",
    "    end_date = datetime(2024, 9, 30)\n",
    "    \n",
    "    for _ in range(num_new):\n",
    "        user_id = random.choice(list(existing_user_ids))\n",
    "        product_id = random.choice(product_ids)\n",
    "        \n",
    "        behavior_date = start_date + timedelta(days=random.randint(0, (end_date - start_date).days))\n",
    "        \n",
    "        steps = []\n",
    "        now = behavior_date\n",
    "        gap = timedelta(minutes=random.randint(30, 120))\n",
    "        \n",
    "        steps.append((\"浏览\", now))\n",
    "        now += gap\n",
    "        \n",
    "        if random.random() < 0.5:\n",
    "            steps.append((\"收藏\", now))\n",
    "            now += gap\n",
    "        \n",
    "        if random.random() < 0.6:\n",
    "            steps.append((\"加购\", now))\n",
    "            now += gap\n",
    "        \n",
    "        buy_prob = 0.2 if random.random() < 0.5 else 0.4\n",
    "        if random.random() < buy_prob:\n",
    "            steps.append((\"购买\", now))\n",
    "        \n",
    "        payment_methods = [\"信用卡\", \"支付宝\", \"储蓄卡\", \"微信\", \"Unknown\"]\n",
    "        for action, time in steps:\n",
    "            behavior_data[\"用户ID\"].append(user_id)\n",
    "            behavior_data[\"商品ID\"].append(product_id)\n",
    "            behavior_data[\"时间\"].append(time.strftime(\"%Y-%m-%d %H:%M:%S\"))\n",
    "            behavior_data[\"行为\"].append(action)\n",
    "            behavior_data[\"备注\"].append(random.choice(payment_methods) if action == \"购买\" else None)\n",
    "    \n",
    "    return pd.DataFrame(behavior_data)\n",
    "\n",
    "# --------------------------\n",
    "# 5. 多线程更新控制逻辑\n",
    "# --------------------------\n",
    "TABLE_GENERATORS = {\n",
    "    \"customers_info\": generate_customers_info,\n",
    "    \"user_behavior_demo\": generate_user_behavior  # 仅修改表名为user_behavior_demo\n",
    "}\n",
    "\n",
    "UPDATE_CONFIG = {\n",
    "    \"customers_info\": {\"interval\": 1, \"num\": 10},\n",
    "    \"user_behavior_demo\": {\"interval\": 1, \"num\": 15}  # 仅修改表名为user_behavior_demo\n",
    "}\n",
    "\n",
    "pause_event = threading.Event()\n",
    "pause_event.set()\n",
    "\n",
    "def update_single_table(table_name, generator, interval, num_per_update):\n",
    "    print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 启动 {table_name} 表更新线程，间隔 {interval} 秒，每次新增 {num_per_update} 条数据\")\n",
    "    retry_count = 0\n",
    "    while True:\n",
    "        if not pause_event.is_set():\n",
    "            print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {table_name} 表暂停，等待恢复...\")\n",
    "            pause_event.wait()\n",
    "            print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {table_name} 表恢复更新\")\n",
    "        try:\n",
    "            start_time = time.time()\n",
    "            new_data_df = generator(num_new=num_per_update)\n",
    "            \n",
    "            new_data_df.to_sql(\n",
    "                name=table_name,\n",
    "                con=engine,\n",
    "                if_exists=\"append\",\n",
    "                index=False,\n",
    "                chunksize=num_per_update,\n",
    "                method='multi'\n",
    "            )\n",
    "            \n",
    "            total_count = pd.read_sql(f\"SELECT COUNT(*) FROM {table_name}\", con=engine).iloc[0, 0]\n",
    "            print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {table_name} 表新增 {num_per_update} 条数据，当前总数据量: {total_count} 条\")\n",
    "            retry_count = 0\n",
    "            elapsed_time = time.time() - start_time\n",
    "            time.sleep(max(0, interval - elapsed_time))\n",
    "        except Exception as e:\n",
    "            retry_count += 1\n",
    "            print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {table_name} 表更新失败，第 {retry_count} 次重试，错误: {str(e)}\")\n",
    "            time.sleep(5)\n",
    "\n",
    "def start_all_table_updates():\n",
    "    threads = []\n",
    "    for table_name, generator in TABLE_GENERATORS.items():\n",
    "        config = UPDATE_CONFIG[table_name]\n",
    "        thread = threading.Thread(\n",
    "            target=update_single_table,\n",
    "            args=(table_name, generator, config[\"interval\"], config[\"num\"]),\n",
    "            daemon=True\n",
    "        )\n",
    "        threads.append(thread)\n",
    "        thread.start()\n",
    "    return threads\n",
    "\n",
    "def toggle_pause():\n",
    "    if pause_event.is_set():\n",
    "        pause_event.clear()\n",
    "        print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 所有表暂停更新\")\n",
    "    else:\n",
    "        pause_event.set()\n",
    "        print(f\"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 所有表恢复更新\")\n",
    "\n",
    "# --------------------------\n",
    "# 6. 程序入口\n",
    "# --------------------------\n",
    "if __name__ == \"__main__\":\n",
    "    print(\"===== 启动用户信息和行为表持续更新程序 =====\")\n",
    "    print(f\"目标数据库: {DB_CONFIG['database']}\")\n",
    "    print(\"提示: 按 Ctrl+C 可停止所有表的更新操作，输入 'pause' 暂停，'resume' 恢复\\n\")\n",
    "    \n",
    "    print(\"✅ 准备在现有表上追加数据，不会创建或替换原有表\")\n",
    "    \n",
    "    update_threads = start_all_table_updates()\n",
    "    try:\n",
    "        while True:\n",
    "            command = input(\"> \")\n",
    "            if command.lower() in [\"pause\", \"resume\"]:\n",
    "                toggle_pause()\n",
    "            else:\n",
    "                print(\"无效命令！支持 'pause'（暂停） / 'resume'（恢复）\")\n",
    "    except KeyboardInterrupt:\n",
    "        print(\"\\n所有表更新线程已手动停止\")\n",
    "    finally:\n",
    "        engine.dispose()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "33dd31c0-0aed-41a8-981a-094fc5c925ac",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.20"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
